package com.atguigu.gmall.common.config.mq.service;

import com.alibaba.fastjson.JSON;
import com.atguigu.gmall.common.constant.RedisConst;
import com.atguigu.gmall.common.util.MD5;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.lang.Nullable;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * @Description
 * @Return
 * @Author https://weiranyi.github.io
 * @Date Created in 2022-12-22 17:29
 **/
@Slf4j
@Service
public class MqService {
    private RabbitTemplate rabbitTemplate;
    private StringRedisTemplate stringRedisTemplate;

    // 如果一个组件只有一个有参构造器；容器创建这个组件对象的时候，就会将所有参数去容器中确定
    public MqService(RabbitTemplate rabbitTemplate, StringRedisTemplate stringRedisTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        this.stringRedisTemplate = stringRedisTemplate;
        initTemplate();
    }

    public <T> T getMessageContent(Message message, Class<T> type) {
        byte[] body = message.getBody();
        String json = new String(body);
        T t = JSON.parseObject(json, type);
        return t;
    }

    private void initTemplate() {
        // 1.成功或失败都响应
        rabbitTemplate.setConfirmCallback((@Nullable CorrelationData correlationData, boolean ack, @Nullable String cause) -> {
            // ack是true说明交换机获取到消息了
            log.info("消息：ack=[{}],原因=[{}],消息的唯一ID&correlationData=[{}]", ack, cause, correlationData);
        });
        // 2.只响应失败的 回复状态码(replyCode)：312 没用路由（NO_ROUTE） 错误消息可以存储到数据库，人工处理
        rabbitTemplate.setReturnCallback((Message message, int replyCode, String replyText,
                                          String exchange, String routingKey) -> {
            log.info("消息：[{}],回复状态码：replyCode=[{}],回复的文本：replyText=[{}],exchange=[{}],routingKey=[{}]", message, replyCode, replyText, exchange, routingKey);
            log.error("消息发送错误！已录入数据库：等待人工确认处理；");
        });

        // 3.还可以设置重试模板
        RetryTemplate retryTemplate = new RetryTemplate();
        rabbitTemplate.setRetryTemplate(retryTemplate);
    }

    public void convertAndSend(String exchange, String routingKey, Object obj) {
        String msg = "";
        if (!(obj instanceof String)) {
            msg = JSON.toJSONString(obj);
        } else {
            msg = obj.toString();
        }
        // 消息的唯一关联
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        // 单独为某一个消息添加回调 correlationData.getFuture().addCallback();
        rabbitTemplate.convertAndSend(exchange, routingKey, msg, correlationData);
    }

    public void retry(Message message, Channel channel, Long count) {
        // redis：计数-统计多少次
        // 计算消息内容的MD5码值，只要内容不变，MD5码值不变=》同数学建模比赛提交资料的资料唯一标识码等
        String json = new String(message.getBody());
        String messageMd5 = MD5.encrypt(json);


        Long aLong = stringRedisTemplate.opsForValue().increment(RedisConst.MQ_RETRY_COUNT + messageMd5);
        if (aLong > count) {
            log.error("消息重试机制超过限制；正在入库，等待人工处理......{}", json);
            // 用完后删除
            stringRedisTemplate.delete(RedisConst.MQ_RETRY_COUNT + messageMd5);
        } else {
            long tag = message.getMessageProperties().getDeliveryTag();
            // 消息回到队列
            try {
                channel.basicNack(tag, false, true);
                stringRedisTemplate.expire(RedisConst.MQ_RETRY_COUNT + messageMd5, 10, TimeUnit.MINUTES);
            } catch (IOException e) {
                // 这发生异常代表网络中断了
            }
        }

    }
}
